use futures_channel::mpsc::{self, UnboundedReceiver, UnboundedSender};
use futures_util::{
    future::{self, Either},
    SinkExt, StreamExt,
};

use crate::utils::ids::id_generate_xbit;
use crate::utils::time_utils;
use std::collections::HashMap;
use std::convert::TryFrom;
use std::sync::Arc;
use tokio::{
    sync::{watch::Receiver as WReceiver, RwLock},
    task::JoinHandle,
};
use tonic::{
    codegen::InterceptedService,
    service::Interceptor,
    transport::{Channel, Endpoint, Server, Uri},
    Status,
};
use tonic_inf::api::{
    byte_buffer_msg::{self},
    send_content_msg::{self},
    tunnel_service_client::TunnelServiceClient,
    tunnel_service_server::TunnelServiceServer,
    BrowserRequest, ByteBufferData, ByteBufferRes, ByteBufferStart, ByteBufferStop,
    SendContentData, SendContentRes, SendContentStart, SendContentStop,
};

use crate::imp::tonicrpc::kcptunnel::{
    kcp::KcpTunnelImp,
    tunnel::{
        TonicServerStopKind, TunnelClientCreator, TunnelConfig, TunnelConn, TunnelConnectMsg,
        TunnelCreatorImp, TunnelPair, TunnelPairWithIp, TunnelStatus, URIInfo,
    },
};
use crate::imp::tonicrpc::tonicserverimp::TonicServerImp;
use crate::imp::tonicrpc::tonictunnel::TonicTunnelCallback;

use tower::service_fn;
use webrtc_ice::state::ConnectionState;

lazy_static! {
    // tonic server-start flag
    static ref TONIC_SERVER_START: state::Storage<RwLock<bool>> = state::Storage::new();

    // kcp with ice map ,key = tunnel_id,value = KcpTunnelImp
    static ref KCP_TUNNEL_MAP: RwLock<HashMap<String, Result<KcpTunnelImp, anyhow::Error>>> =
    RwLock::new(HashMap::new());

    // tonic tunnl creator watch, send and received TunnelCreatorImp
    static ref TUNNEL_CREATOR_WATCH: (
        // sender and receiver only contains the lastest TunnelCreatorImp.
        tokio::sync::watch::Sender<Option<TunnelCreatorImp>>,
        tokio::sync::watch::Receiver<Option<TunnelCreatorImp>>,
    ) = tokio::sync::watch::channel(None);

    //  tonic client sender map. key = tunnel_id,value = TonicClientCommand
    static ref TUNNEL_CLIENT_COMMAND_SENDER_MAP: std::sync::RwLock<HashMap<String, UnboundedSender<TunnelClientCommand>>> =
        std::sync::RwLock::new(HashMap::new());

    // remote content callback. option tokio::sync::mpsc::Sender
    static ref REMOTE_CONTENT_CALLBACK: RwLock<Option<tokio::sync::mpsc::UnboundedSender<send_content_msg::Msg>>> =
        RwLock::new(None);

    // remote content service sender map.
    static ref REMOTE_CONTENT_SENDER_MAP: RwLock<HashMap<String, (UnboundedSender<SendContentCommand>, String, String)>> =
        RwLock::new(HashMap::new());

    // remote bytebuffer callback. option tokio::sync::mpsc::Sender
    static ref REMOTE_BYTEBUFFER_CALLBACK: RwLock<Option<tokio::sync::mpsc::UnboundedSender<byte_buffer_msg::Bfmsg>>> =
        RwLock::new(None);

    // remote bytebuffer service sender map.
    static ref REMOTE_BYTEBUFFER_SENDER_MAP: RwLock<HashMap<String, (UnboundedSender<SendByteBufferCommand>, String, String)>> =
        RwLock::new(HashMap::new());

    // tunnel service
    // map key = tunnel_id, value = tuple(TunnelActionCommand sender,local client id,remote client id)
    static ref TUNNEL_SERVICE_COMMAND_SENDER_MAP: std::sync::RwLock<HashMap<String, (UnboundedSender<TunnelServiceCommand>, String, String)>> =
        std::sync::RwLock::new(HashMap::new());
}

// check if tonic sevrer is started
pub fn is_tonic_server_start() -> bool {
    let service_start = if let Some(service_start) = TONIC_SERVER_START.try_get() {
        if let Ok(service_start) = service_start.try_read() {
            *service_start
        } else {
            false
        }
    } else {
        false
    };
    log::info!("is tonic serivce start: {}", service_start);
    service_start
}

// set tonic server started flag
pub fn set_tonic_server_start(start: bool) -> bool {
    let ok = if let Some(service_start) = TONIC_SERVER_START.try_get() {
        if let Ok(mut service_start) = service_start.try_write() {
            if *service_start != start {
                *service_start = start;
                true
            } else {
                false
            }
        } else {
            false
        }
    } else {
        TONIC_SERVER_START.set(RwLock::new(start));
        true
    };
    return ok;
}

/// tonic server
pub struct TonicServer {
    // kcp tunnel future handler
    tunnel_handler: JoinHandle<Result<(), anyhow::Error>>,
    // tonic server future handler
    service_handler: JoinHandle<Result<(), anyhow::Error>>,
    // flag r is ok.
    is_ok: Arc<RwLock<bool>>,
}

impl TonicServer {
    // create a tonic server with kcp tunnel.
    pub async fn create_service_with_cb(
        // local client id
        client_id: String,
        // tonic callback
        mut callback: Box<dyn TonicTunnelCallback>,
        // config store path
        config_store_path: String,
        // turn uri
        turn_uri: URIInfo,
        // download file saved path
        download_base_path: String,
        // http download size callback
        enable_http_callback: bool,
    ) -> Result<Self, anyhow::Error> {
        // create a futures_channel mpsc unbound tx/rx pair
        let (service_callback_tx, mut service_callback_rx) =
            futures_channel::mpsc::unbounded::<TunnelConnectMsg>();

        // general tunnel options
        let options = TunnelConfig {
            config_store_path,
            client_id: client_id.to_owned(),
            turn_uri,
            download_base_path,
            enable_http_callback,
        };
        // set if enable http call back
        let enable_http_callback = options.enable_http_callback;

        // set call back hander = tokio spwan future.
        // spwan in async function.
        tokio::spawn(async move {
            // create a tokio mpsc channel message
            let (on_remote_content_tx, mut on_remote_content_rx) =
                tokio::sync::mpsc::unbounded_channel::<send_content_msg::Msg>();

            let (on_byte_buffer_tx, mut on_byte_buffer_rx) =
                tokio::sync::mpsc::unbounded_channel::<byte_buffer_msg::Bfmsg>();

            // set remote call back handler
            get_or_set_remote_content_callback(true, Some(on_remote_content_tx)).await;

            // set bytebuffer call back handler
            get_or_set_bytebuffer_callback(true, Some(on_byte_buffer_tx)).await;

            // run here!
            let mut pre_call_back_time = tokio::time::Instant::now();
            loop {
                // select branch, if match service callback
                log::info!("create_service_with_cb tokio::spawn callback loop run.");
                tokio::select! {
                    // one branch loop for service_callback_rx
                    msg = service_callback_rx.next() => match msg {
                        Some(msg) => {
                            msg.on_connect_callback(&mut callback, enable_http_callback);
                        },
                        None => {
                            log::error!("callback_handler break select! callback!");
                            break;
                        }
                    },
                    // one branch loop for on_remote_content_rx
                    msg = on_remote_content_rx.recv() =>{
                        log::info!("on_remote_content_rx.recv()");
                        match msg {
                            Some(msg) => {
                                let current = tokio::time::Instant::now();
                                let duration = current.duration_since(pre_call_back_time);
                                on_remote_content_callback(msg, &mut callback);
                                pre_call_back_time = tokio::time::Instant::now();
                                log::info!("on_remote_content_rx.recv() callback cost:{:?}",duration);
                            },
                            None => {
                                log::error!("callback_handler break select! callback!");
                                break;
                            }
                        }
                    },

                    bfmsg = on_byte_buffer_rx.recv()=> {
                        log::info!("on_byte_buffer_rx.recv()");
                        match bfmsg {
                            Some(msg) => {
                                let current = tokio::time::Instant::now();
                                let duration = current.duration_since(pre_call_back_time);
                                on_byte_buffer_callback(msg, &mut callback);
                                pre_call_back_time = tokio::time::Instant::now();
                                log::info!("on_byte_buffer_rx.rec() callback cost:{:?}",duration);
                            },
                            None => {
                                log::error!("callback_handler break select! callback!");
                                break;
                            }
                        }
                    }
                }
            }
        });

        // create server
        let server = TonicServer::new(&client_id, service_callback_tx, &options).await?;
        // server.callback_handler = Some(callback_handler);
        Ok(server)
    }

    // create a tunnel handle and a tonic server handler
    async fn new(
        client_id: &str,
        callback_tx: UnboundedSender<TunnelConnectMsg>,
        options: &TunnelConfig,
    ) -> Result<Self, anyhow::Error> {
        let is_ok = Arc::new(RwLock::new(true));
        let device_id = client_id.to_owned();

        // create tunnel_tx and tunnel_rx
        let (tunnel_tx, tunnel_rx) = mpsc::unbounded::<TunnelConnectMsg>();
        let options = options.to_owned();

        // start kcp agent based on webrtc tunnel
        let tunnel_is_ok = is_ok.clone();
        let tunnel_handler = tokio::spawn(async move {
            let res = TunnelCreatorImp::start_tonic_tunnel(&device_id, tunnel_tx, &options).await;
            *tunnel_is_ok.write().await = false;
            res
        });

        // start tonic rpc server
        let service_is_ok = is_ok.clone();
        let device_id = client_id.to_owned();
        let service_handler = tokio::spawn(async move {
            let res = TonicServer::start_tonic_server(&device_id, tunnel_rx, callback_tx).await;
            *service_is_ok.write().await = false;
            res
        });

        // return tonic server struct
        Ok(TonicServer {
            tunnel_handler,
            service_handler,
            is_ok,
            // callback_handler: None,
        })
    }

    pub async fn is_ok(&self) -> bool {
        *self.is_ok.read().await
    }

    pub(crate) async fn join(self) -> Result<(), anyhow::Error> {
        // wait for left or right end!
        let res = match future::select(self.service_handler, self.tunnel_handler).await {
            Either::Left((res, _)) => res?,
            Either::Right((res, _)) => res?,
        };
        // release content tunnel.
        get_or_set_remote_content_callback(true, None).await;

        // release bytebuffer tunnel.
        get_or_set_bytebuffer_callback(true, None).await;

        log::info!("join exit async_start_tonic_server!");

        res
    }

    // async start tonic server
    async fn start_tonic_server(
        client_id: &str,
        mut tunnel_rx: UnboundedReceiver<TunnelConnectMsg>,
        callback_tx: UnboundedSender<TunnelConnectMsg>,
    ) -> Result<(), anyhow::Error> {
        log::info!("start_tonic_server my_cleint_id: {}", client_id);

        let tunnel_service = TonicServerImp {};
        // let addr = "[::1]:10000".parse().unwrap();

        let incoming = {
            // streaming from tunnel_rx
            async_stream::stream! {
              use TunnelConnectMsg::*;
              // received message from kcp tunnel
              while let Some(msg) = tunnel_rx.next().await {
                  //  callback tx send tunnel ice message

                match callback_tx.unbounded_send(msg.to_owned()){
                    Ok(()) => {},
                    Err(e) => {
                        if e.is_disconnected() {
                            log::error!("start_tonic_server callback_tx.unbounded_send close error.break!");
                            break;
                        }
                        if e.is_full() {
                            log::error!("start_tonic_server callback_tx.unbounded_send full error ,wait...");
                        }
                    }
                }
                // if let Err(e) =  {
                //     log::error!("start_tonic_server callback_tx.unbounded_send err: {:?}", e);

                // };
                match msg {
                    // log tunnel destroyed!
                    OnTunnelConnectDestroy{tunnel_id,local_client_id, remote_client_id} => {
                        log::info!("tunnel OnTunnelConnectDestroy tunnel_id:{:?} local_client_id:{:?} remote_device_id:{:?}", tunnel_id,local_client_id, remote_client_id);
                    },
                    // log tunnel pair channged!
                    OnTunnelPairChange(msg) => {
                        log::info!("tunnel OnTunnelPairChange msg:{:?}", msg);
                    },
                    // log tunnel connect ready!
                    OnTunnelConnectReady(conn) => {
                        log::info!("tunnel OnTunnelConnectReady: {:?}", conn);
                        let is_controlling = conn.is_controlling;
                        let tunnel_id = conn.tunnel_id.to_owned();
                        let kcp_tunnel = conn.to_kcp_tunnel().await;

                        if !is_controlling {
                            // let res: Result<KcpTunnelImp, anyhow::Error> = Ok(conn);
                            log::info!("tunnel OnTunnelConnectReady is_controlling false");
                            // kcp tunnel return to tonic,set as a service!
                            // res implementation of tokio read and write
                            yield kcp_tunnel;
                        } else {
                            // if iscontrolling.
                            log::info!("tunnel OnTunnelConnectReady is_controlling true");

                            // set to kcp map
                            match set_or_take_kcp(&tunnel_id, Some(kcp_tunnel)).await {
                                Ok(_item) => {
                                    log::info!("tunnel OnTunnelConnectReady set_or_take_kcp success!");
                                },
                                Err(e) => {
                                    log::error!("set_or_take_kcp err: {:?}", e);
                                }
                            }
                        }
                    },
                    OnTonicServerCreated(creator) => {
                        log::info!("tunnel OnTonicServerCreated: {:?}", &creator);
                        if let Err(e) = set_tunnel_creator(Some(creator)) {
                            log::error!("OnTunnelServiceCreated:: set_tunnel_creator err: {:?}", e);
                        }
                    },
                    OnTonicServerDroped(kind) => {
                        log::info!("tunnel OnTonicServerDroped: {:?}", &kind);
                        if let Err(e) = clear_tonic_tunnel_map().await {
                            log::info!("clear_kcp_map err: {:?}", &e);
                        }
                        if let Err(e) = set_tunnel_creator(None) {
                            log::info!("set_tunnel_creator err: {:?}", &e);
                        }
                        yield Err(anyhow::anyhow!("tunnel OnServiceStop: {:?}", &kind));
                    },
                    OnTunnelStateChange(state) => {
                        log::info!("tunnel OnTunnelStateChange: {:?}", state);
                        let need_close = match state.state {
                            ConnectionState::Closed => {
                                true
                            }
                            ConnectionState::Failed => {
                                true
                            },
                            _ => {
                                false
                            }
                        };
                        if need_close {
                            if let Err(e) = close_remote_tonic_tunnel(state.tunnel_id.to_owned()) {
                                log::error!("ConnectionState::Closed close_remote_service tunnel_id: {} err: {:?}", &state.tunnel_id, e);
                            }
                        }
                    }
                }
              }
            }
        };

        Server::builder()
            .add_service(TunnelServiceServer::new(tunnel_service))
            // .serve(addr)
            .serve_with_incoming(incoming)
            .await?;
        Ok(())
    }
}

// stop tonic client command
pub enum TunnelClientCommand {
    StopConnect,
    StopByteBuffer,
}

#[derive(Debug, Clone)]
pub enum SendContentCommand {
    // send content cmd
    SendContent(SendContentData),
    // start content service cmd
    Start(SendContentStart),
    // stop content service cmd
    Stop(SendContentStop),
}

#[derive(Debug, Clone)]
pub enum SendByteBufferCommand {
    // send byte buffer cmd
    SendByteBuffer(ByteBufferData),
    // start buffer buffer service cmd
    Start(ByteBufferStart),
    // stop buffer buffer service cmd
    Stop(ByteBufferStop),
}

// // download message
// pub struct DownloadMsg {
//     // task id
//     pub task_id: String,
//     // saved file path
//     pub save_file_path: String,
//     // download file path
//     pub download_file_path: String,
// }

// tunnel service command
pub enum TunnelServiceCommand {
    // Download(DownloadMsg),
    // BrowseFile((String, BrowserRequest)),
    SendContentStart((SendContentStart, i64)),
    SendByteBufferStart((ByteBufferStart, i64)),
}

// map key = tunnel_id, value = tuple(TunnelActionCommand sender,local client id,remote client id)
pub fn get_or_set_service_command_sender(
    is_set: bool,
    tunnel_id: &str,
    client: Option<(UnboundedSender<TunnelServiceCommand>, String, String)>,
) -> Result<Option<(UnboundedSender<TunnelServiceCommand>, String, String)>, anyhow::Error> {
    log::info!(
        "get_or_set_service_command_sender is_set: {} tunnel_id: {} client.is_some: {}",
        is_set,
        tunnel_id,
        client.is_some()
    );
    let res = if is_set {
        let mut map = TUNNEL_SERVICE_COMMAND_SENDER_MAP
            .try_write()
            .map_err(|e| anyhow::anyhow!("map.try_write() err: {:?} ", e))?;

        let res = match client {
            Some(client) => map.insert(tunnel_id.to_owned(), client),
            None => map.remove(tunnel_id),
        };
        res
    } else {
        match TUNNEL_SERVICE_COMMAND_SENDER_MAP
            .try_read()
            .map_err(|e| anyhow::anyhow!("map.try_read() err: {:?} ", e))?
            .get(tunnel_id)
        {
            Some(client) => Some(client.clone()),
            None => None,
        }
    };

    Ok(res)
}

// find Kcp Impl from map,if kcp null ,then set to Some(Err) Option
pub async fn if_no_kcp_set_err(tunnel_id: &str, err: anyhow::Error) {
    if !has_kcp_ok(tunnel_id).await {
        if let Err(e) = set_or_take_kcp(tunnel_id, Some(Err(err))).await {
            log::info!("if_no_kcp_set_err set_or_take_kcp e: {}", &e);
        }
    }
}

// find kcp in map
pub async fn has_kcp_ok(tunnel_id: &str) -> bool {
    let map = KCP_TUNNEL_MAP.read().await;
    let res = match map.get(tunnel_id) {
        Some(item) => item.is_ok(),
        None => false,
    };
    log::info!("has_kcp_ok {}", res);
    res
}

// set or take kcp
async fn set_or_take_kcp(
    tunnel_id: &str,
    tunnel: Option<Result<KcpTunnelImp, anyhow::Error>>,
) -> Result<Option<Result<KcpTunnelImp, anyhow::Error>>, anyhow::Error> {
    let mut map = KCP_TUNNEL_MAP.write().await;
    let res = match tunnel {
        // tunnel may be error!
        Some(tunnel) => map.insert(tunnel_id.to_owned(), tunnel),
        None => map.remove(tunnel_id),
    };
    Ok(res)
}

// clear all kcp map.
async fn clear_tonic_tunnel_map() -> Result<(), anyhow::Error> {
    log::info!("clear_kcp_map");
    let tunnels: Vec<String> = {
        let map = TUNNEL_CLIENT_COMMAND_SENDER_MAP
            .try_read()
            .map_err(|e| anyhow::anyhow!(e.to_string()))?;
        map.keys().map(|k| k.to_owned()).collect()
    };
    for tunnel in tunnels {
        if let Err(e) = close_remote_tonic_tunnel(tunnel) {
            log::error!("clear_kcp_map close_remote_service err: {:?}", e);
        }
    }
    let mut map = KCP_TUNNEL_MAP.write().await;
    map.clear();
    Ok(())
}

// get or set remote content callback.
pub async fn get_or_set_remote_content_callback(
    is_set: bool,
    new: Option<tokio::sync::mpsc::UnboundedSender<send_content_msg::Msg>>,
) -> Option<tokio::sync::mpsc::UnboundedSender<send_content_msg::Msg>> {
    if is_set {
        let mut val = REMOTE_CONTENT_CALLBACK.write().await;
        let res = val.take();
        *val = new;
        res
    } else {
        REMOTE_CONTENT_CALLBACK.read().await.clone()
    }
}

// get or set content service sender
pub fn get_or_set_content_sender(
    is_set: bool,
    tunnel_id: &str,
    sender: Option<(UnboundedSender<SendContentCommand>, String, String)>,
) -> Result<Option<(UnboundedSender<SendContentCommand>, String, String)>, anyhow::Error> {
    log::info!(
        "get_or_set_content_sender is_set: {} tunnel_id: {} client.is_some: {}",
        is_set,
        tunnel_id,
        sender.is_some()
    );
    anyhow::ensure!(tunnel_id != "", "tunnel id empty str");
    let res = if is_set {
        let mut map = REMOTE_CONTENT_SENDER_MAP
            .try_write()
            .map_err(|e| anyhow::anyhow!("map.try_write() err: {:?} ", e))?;
        let res = match sender {
            Some(sender) => map.insert(tunnel_id.to_owned(), sender),
            None => map.remove(tunnel_id),
        };
        res
    } else {
        match REMOTE_CONTENT_SENDER_MAP
            .try_read()
            .map_err(|e| anyhow::anyhow!("map.try_read() err: {:?} ", e))?
            .get(tunnel_id)
        {
            Some(client) => Some(client.clone()),
            None => None,
        }
    };
    Ok(res)
}

// select content service duplex implementation ()
pub fn select_content_service_duplex(tunnel_id: String) -> Result<(), anyhow::Error> {
    let (sender, my_device_id, remote_device_id) =
        get_or_set_service_command_sender(false, &tunnel_id, None)?.ok_or(anyhow::anyhow!(
            "start_send_content_service for tunnel_id:{} not connect",
            tunnel_id
        ))?;

    sender.unbounded_send(TunnelServiceCommand::SendContentStart((
        SendContentStart {
            tunnel_id,
            my_device_id,
            remote_device_id,
            is_controlling: true,
        },
        time_utils::timestamp_ms_i64(),
    )))?;
    Ok(())
}

// stop tonic send content service implementation ()
pub fn cancel_content_service_duplex(
    tunnel_id: String,
    is_confirm: bool,
) -> Result<(), anyhow::Error> {
    let (sender, _my_device_id, _remote_device_id) =
        get_or_set_content_sender(false, &tunnel_id, None)?.ok_or(anyhow::anyhow!(
            "already stoped or not call start_send_content_service with this tunnel id: {}",
            tunnel_id
        ))?;

    sender.unbounded_send(SendContentCommand::Stop(SendContentStop {
        tunnel_id,
        is_confirm,
    }))?;
    Ok(())
}

// get or set remote bytebuffer callback.
pub async fn get_or_set_bytebuffer_callback(
    is_set: bool,
    new: Option<tokio::sync::mpsc::UnboundedSender<byte_buffer_msg::Bfmsg>>,
) -> Option<tokio::sync::mpsc::UnboundedSender<byte_buffer_msg::Bfmsg>> {
    if is_set {
        let mut val = REMOTE_BYTEBUFFER_CALLBACK.write().await;
        let res = val.take();
        *val = new;
        res
    } else {
        REMOTE_BYTEBUFFER_CALLBACK.read().await.clone()
    }
}

// get or set bytebuffer service sender
pub fn get_or_set_bytebuffer_sender(
    is_set: bool,
    tunnel_id: &str,
    sender: Option<(UnboundedSender<SendByteBufferCommand>, String, String)>,
) -> Result<Option<(UnboundedSender<SendByteBufferCommand>, String, String)>, anyhow::Error> {
    log::info!(
        "get_or_set_bytebuffer_sender is_set: {} tunnel_id: {} client.is_some: {}",
        is_set,
        tunnel_id,
        sender.is_some()
    );
    anyhow::ensure!(tunnel_id != "", "tunnel id empty str");
    let res = if is_set {
        let mut map = REMOTE_BYTEBUFFER_SENDER_MAP
            .try_write()
            .map_err(|e| anyhow::anyhow!("remote_bytebuf_map.try_write() err: {:?} ", e))?;
        let res = match sender {
            Some(sender) => map.insert(tunnel_id.to_owned(), sender),
            None => map.remove(tunnel_id),
        };
        res
    } else {
        match REMOTE_BYTEBUFFER_SENDER_MAP
            .try_read()
            .map_err(|e| anyhow::anyhow!("remote_bytebuf_map.try_read() err: {:?} ", e))?
            .get(tunnel_id)
        {
            Some(client) => Some(client.clone()),
            None => None,
        }
    };
    Ok(res)
}

pub fn select_bytebuf_service_duplex(tunnel_id: String) -> Result<(), anyhow::Error> {
    let (sender, my_device_id, remote_device_id) =
        get_or_set_service_command_sender(false, &tunnel_id, None)?.ok_or(anyhow::anyhow!(
            "start_send_bytebuf_service for tunnel_id:{} not connect",
            tunnel_id
        ))?;

    sender.unbounded_send(TunnelServiceCommand::SendByteBufferStart((
        ByteBufferStart {
            tunnel_id,
            my_device_id,
            remote_device_id,
            is_controlling: true,
        },
        time_utils::timestamp_ms_i64(),
    )))?;
    Ok(())
}

pub fn cancel_bytebuf_service_duplex(
    tunnel_id: String,
    is_confirm: bool,
) -> Result<(), anyhow::Error> {
    let (sender, _my_device_id, _remote_device_id) =
        get_or_set_bytebuffer_sender(false, &tunnel_id, None)?.ok_or(anyhow::anyhow!(
            "already stoped or not call start_send_bytebuf_service with this tunnel id: {}",
            tunnel_id
        ))?;

    sender.unbounded_send(SendByteBufferCommand::Stop(ByteBufferStop {
        tunnel_id,
        is_confirm,
    }))?;

    Ok(())
}

// not usage
// pub fn request_download_file(
//     tunnel_id: String,
//     save_file_path: String,
//     download_file_path: String,
// ) -> Result<String, anyhow::Error> {
//     let (sender, _, _) =
//         get_or_set_service_command_sender(false, &tunnel_id, None)?.ok_or(anyhow::anyhow!(
//             "request_download_file for tunnel_id:{} not connect",
//             tunnel_id
//         ))?;

//     let task_id = id_generate_xbit(4);
//     sender.unbounded_send(TunnelServiceCommand::Download(DownloadMsg {
//         task_id: task_id.clone(),
//         download_file_path,
//         save_file_path,
//     }))?;

//     Ok(task_id)
// }

// pub fn request_browse_file(
//     tunnel_id: String,
//     browse_path: String,
//     need_sha256: bool,
//     goto_folder: String,
// ) -> Result<String, anyhow::Error> {
//     let (sender, _, _) =
//         get_or_set_service_command_sender(false, &tunnel_id, None)?.ok_or(anyhow::anyhow!(
//             "request_browse_file for tunnel_id:{} not connect",
//             tunnel_id
//         ))?;

//     let task_id = id_generate_xbit(4);
//     let request = BrowserRequest {
//         relative_path: browse_path,
//         need_sha256,
//         goto_folder,
//     };
//     sender.unbounded_send(TunnelServiceCommand::BrowseFile((task_id.clone(), request)))?;

//     Ok(task_id)
// }

pub fn request_send_content(
    tunnel_id: String,
    id: String,
    event: String,
    data: String,
    can_drop: bool,
) -> Result<String, anyhow::Error> {
    let (sender, my_device_id, remote_device_id) =
        get_or_set_content_sender(false, &tunnel_id, None)?.ok_or(anyhow::anyhow!(
            "not call start_send_content_service with this tunnel id: {}",
            tunnel_id
        ))?;
    let id = if id == "" { id_generate_xbit(4) } else { id };
    sender.unbounded_send(SendContentCommand::SendContent(SendContentData {
        tunnel_id,
        my_device_id,
        remote_device_id,
        id: id.to_owned(),
        event,
        data,
        can_drop: Some(can_drop),
    }))?;
    Ok(id)
}

pub fn request_send_bytes(
    tunnel_id: String,
    id: String,
    can_drop: bool,
    video_buf: Vec<u8>,
    audio_buf: Vec<u8>,
) -> Result<String, anyhow::Error> {
    let (sender, _, _) =
        get_or_set_bytebuffer_sender(false, &tunnel_id, None)?.ok_or(anyhow::anyhow!(
            "not call start_send_byte_buffer with this tunnel id: {}",
            tunnel_id
        ))?;
    let id = if id == "" { id_generate_xbit(4) } else { id };
    log::info!(
        "request_send_bytes video len = {:?} audio len = {:?}",
        video_buf.len(),
        audio_buf.len()
    );
    sender.unbounded_send(SendByteBufferCommand::SendByteBuffer(ByteBufferData {
        id: id.to_owned(),
        tunnel_id: tunnel_id.to_owned(),
        video_buffer: video_buf.to_owned(),
        audio_buffer: audio_buf.to_owned(),
        can_drop: Some(can_drop),
    }))?;
    Ok(id)
}

pub fn request_gather_candidates() -> Result<(), anyhow::Error> {
    log::info!("request_gather_candidates");
    let mut creator = get_tunnel_creator().ok_or(anyhow::anyhow!("service not start"))?;
    creator.gather_candidates()
}

// get or set tunnel client command sender
pub fn get_or_set_tunnel_client_command_sender(
    is_set: bool,
    tunnel_id: &str,
    client: Option<UnboundedSender<TunnelClientCommand>>,
) -> Result<Option<UnboundedSender<TunnelClientCommand>>, anyhow::Error> {
    log::info!(
        "get_or_set_tunnel_command_sender is_set: {} tunnel_id: {} client.is_some: {}",
        is_set,
        tunnel_id,
        client.is_some()
    );
    let res = if is_set {
        let mut map = TUNNEL_CLIENT_COMMAND_SENDER_MAP
            .try_write()
            .map_err(|e| anyhow::anyhow!("map.try_write() err: {:?} ", e))?;
        let res = match client {
            Some(client) => map.insert(tunnel_id.to_owned(), client),
            None => map.remove(tunnel_id),
        };
        res
    } else {
        match TUNNEL_CLIENT_COMMAND_SENDER_MAP
            .try_read()
            .map_err(|e| anyhow::anyhow!("map.try_read() err: {:?} ", e))?
            .get(tunnel_id)
        {
            Some(client) => Some(client.clone()),
            None => None,
        }
    };

    Ok(res)
}

// close remote service with tunnel id
pub fn close_remote_tonic_tunnel(tunnel_id: String) -> Result<(), anyhow::Error> {
    log::info!("close_remote_tonic_tunnel tunnel_id: {}", tunnel_id);
    anyhow::ensure!(tunnel_id != "", "tunnel_id is empty");
    // first close content service
    // if this tunnel is not a send_content service tunnel,then warning
    if let Err(e) = cancel_content_service_duplex(tunnel_id.to_owned(), true) {
        log::warn!(
            "close_remote_tonic_tunnel cancel_content_service_duplex err: {:?}",
            &e
        );
    }

    if let Err(e) = cancel_bytebuf_service_duplex(tunnel_id.to_owned(), true) {
        log::warn!(
            "close_remote_tonic_tunnel cancel_bytebuf_service_duplex err: {:?}",
            &e
        );
    }

    // then close tunnel.
    let sender = get_or_set_tunnel_client_command_sender(false, &tunnel_id, None)?.ok_or(
        anyhow::anyhow!(
            "close_remote_tonic_tunnel for tunnel_id:{} not connect",
            tunnel_id
        ),
    )?;
    sender.unbounded_send(TunnelClientCommand::StopConnect)?;
    Ok(())
}

pub fn set_tunnel_creator(sender_tx: Option<TunnelCreatorImp>) -> Result<(), anyhow::Error> {
    // let lock = SENDER_WATCH.read();
    TUNNEL_CREATOR_WATCH.0.send(sender_tx)?;
    Ok(())
}

pub fn get_tunnel_creator() -> Option<TunnelCreatorImp> {
    TUNNEL_CREATOR_WATCH.1.borrow().clone()
}

// close kcp tunnel connect
pub fn close_tunnel_connect(tunnel_id: &str) {
    log::info!("close_tunnel_connect tunnel_id:{} ", tunnel_id);
    match get_tunnel_creator() {
        Some(mut creator) => {
            let res = creator.disconnect(tunnel_id);
            match res {
                Ok(_) => {
                    log::info!("creator.disconnect tunnel_id: {} Ok", tunnel_id);
                }
                Err(e) => {
                    log::error!("creator.disconnect tunnel_id:{} err: {:?}", tunnel_id, e);
                }
            }
        }
        None => {
            log::error!("get_tunnel_creator None");
        }
    }
}

// remote content callback.
fn on_remote_content_callback(
    msg: send_content_msg::Msg,
    callback: &mut Box<dyn TonicTunnelCallback>,
) {
    match msg {
        send_content_msg::Msg::Data(data) => {
            callback.on_remote_content(data);
        }
        send_content_msg::Msg::Res(res) => {
            callback.on_send_content_result(res);
        }
        send_content_msg::Msg::Start(info) => {
            callback.on_content_service_start(info);
        }
        send_content_msg::Msg::Stop(SendContentStop {
            tunnel_id,
            is_confirm,
        }) => {
            if is_confirm {
                callback.on_content_service_stop(tunnel_id);
            } else {
                log::info!("on_remote_content_callback is_confirm false not callback");
            }
        }
        _ => {}
    }
}

// remote bytebuffer callback.
fn on_byte_buffer_callback(
    msg: byte_buffer_msg::Bfmsg,
    callback: &mut Box<dyn TonicTunnelCallback>,
) {
    match msg {
        byte_buffer_msg::Bfmsg::BfData(data) => {
            callback.on_remote_bufferbuffer(data);
        }
        byte_buffer_msg::Bfmsg::BfRes(res) => {
            callback.on_send_bytebuffer_result(res);
        }
        byte_buffer_msg::Bfmsg::BfStart(start) => {
            callback.on_bytebuffer_service_start(start);
        }
        byte_buffer_msg::Bfmsg::BfStop(ByteBufferStop {
            tunnel_id,
            is_confirm,
        }) => {
            if is_confirm {
                callback.on_bytebuffer_service_stop(tunnel_id);
            } else {
                log::info!("on_bytebuffer_service_stop is_confirm false not callback");
            }
        }
        _ => {}
    }
}

pub type QueryParam<'a> = (&'a str, &'a str);
pub type QueryParams<'a> = Vec<QueryParam<'a>>;
pub fn querify(string: &str) -> QueryParams {
    let mut v = Vec::new();
    for pair in string.split('&') {
        let mut it = pair.split('=').take(2);
        let kv = match (it.next(), it.next()) {
            (Some(k), Some(v)) => (k, v),
            _ => continue,
        };
        v.push(kv);
    }
    v
}

// get tunnel creator watch
pub async fn get_tunnel_creator_watch() -> Result<TunnelCreatorImp, anyhow::Error> {
    anyhow::ensure!(is_tonic_server_start(), "tonic server not start");

    // get crator impl receiver, if service started ,tunnel crator impl is return.
    let mut receiver: WReceiver<Option<TunnelCreatorImp>> = {
        let crator_receiver = TUNNEL_CREATOR_WATCH.1.clone();
        crator_receiver
    };

    //  get crator impl.
    let tunnel_crator_imp = receiver.borrow_and_update().clone();
    // tunnel crator imp
    match tunnel_crator_imp {
        Some(tunnel_crator) => return Ok(tunnel_crator.clone()),
        None => {
            receiver.changed().await?;
            if let Some(tunnel_crator) = receiver.borrow_and_update().clone() {
                return Ok(tunnel_crator);
            } else {
                log::error!("receive.changed still None");
                anyhow::bail!("receive.changed still None");
            }
        }
    }
}

// get kcp imp url
async fn get_kcp_imp(url: Uri) -> Result<KcpTunnelImp, anyhow::Error> {
    log::info!("get_kcp_imp url: {}", &url);
    let querys = querify(url.query().unwrap_or_default());
    let target_id = querys
        .iter()
        .find(|item| item.0 == "target_id")
        .ok_or(anyhow::anyhow!("target_id not get"))?
        .1;
    let tunnel_id = querys
        .iter()
        .find(|item| item.0 == "tunnel_id")
        .ok_or(anyhow::anyhow!("tunnel_id not get"))?
        .1;

    let mut interval = tokio::time::interval(std::time::Duration::from_millis(200));
    let mut has_requrest = false;
    // here loop for set_or_take_kcp have value!
    loop {
        // set none kcp tunnel for KCP_MAP,and take current kcp
        match set_or_take_kcp(tunnel_id, None).await? {
            Some(kcp) => {
                if !has_requrest && kcp.is_err() {
                    continue;
                }
                return Ok(kcp?);
            }
            None => {
                if has_requrest {
                    // if has request than wait interval.tick().
                    interval.tick().await;
                } else {
                    // wait for creator ready!.
                    let mut creator = get_tunnel_creator_watch().await?;
                    creator.request_connect(target_id, tunnel_id).await?;
                    has_requrest = true;
                    interval.tick().await;
                }
            }
        }
    }
}

#[derive(Clone)]
pub struct FlowControlInterceptor {}

impl FlowControlInterceptor {}

impl Interceptor for FlowControlInterceptor {
    // tonic client interceptor
    fn call(&mut self, request: tonic::Request<()>) -> Result<tonic::Request<()>, Status> {
        log::info!("tonic rpc flow control interceptor,not usage for steaming mode!");
        return Ok(request);
    }
}

// create a localhost channel with connector.
pub(crate) async fn _get_tonic_client(
    target_id: &str,
    tunnel_id: &str,
) -> Result<TunnelServiceClient<InterceptedService<Channel, FlowControlInterceptor>>, anyhow::Error>
{
    log::info!("_get_tonic_client");

    // let concurrency_num = num_cpus::get();
    let channel = Endpoint::try_from(format!(
        "http://[::]:0?target_id={}&tunnel_id={}",
        target_id, tunnel_id
    ))?
    .keep_alive_timeout(std::time::Duration::from_secs(20))
    .keep_alive_while_idle(false)
    .connect_timeout(std::time::Duration::from_secs(10)) //10 seconds timeout connect_with_connector
    .timeout(std::time::Duration::from_secs(5))
    .concurrency_limit(2)
    .connect_with_connector(service_fn(|url: Uri| {
        log::info!("_get_tonic_client service_fn url:{}", url);
        let res = async move {
            // service_fn for client implementation.
            let res = get_kcp_imp(url).await;
            log::warn!("_get_tonic_client get_kcp_imp done! res:{:?}", res);
            res
        };
        res
    }))
    .await?;

    let client: TunnelServiceClient<InterceptedService<Channel, FlowControlInterceptor>> =
        TunnelServiceClient::with_interceptor(channel, FlowControlInterceptor {});

    Ok(client)
}
